#pragma once

#include "muduo/net/TcpConnection.h"
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"

#include "../mq_comm/mq_helper.hpp"
#include "../mq_comm/mq_log.hpp"
#include "../mq_comm/mq_msg.pb.h"
#include "../mq_comm/mq_threadpool.hpp"
#include "../mq_comm/mq.pb.h"

#include "mq_host.hpp"
#include "mq_consumer.hpp"
#include "mq_route.hpp"


namespace mq
{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using openChannelRequestPtr = std::shared_ptr<openChannelRequest>;
    using closeChannelRequestPtr = std::shared_ptr<closeChannelRequest>;
    using declareExchangeRequestPtr = std::shared_ptr<declareExchangeRequest>;
    using deleteExchangeRequestPtr = std::shared_ptr<deleteExchangeRequest>;
    using declareQueueRequestPtr = std::shared_ptr<declareQueueRequest>;
    using deleteQueueRequestPtr = std::shared_ptr<deleteQueueRequest>;
    using queueBindRequestPtr = std::shared_ptr<queueBindRequest>;
    using queueUnBindRequestPtr = std::shared_ptr<queueUnBindRequest>;
    using basicPublishRequestPtr = std::shared_ptr<basicPublishRequest>;
    using basicAckRequestPtr = std::shared_ptr<basicAckRequest>;
    using basicConsumeRequestPtr = std::shared_ptr<basicConsumeRequest>;
    using basicCancelRequestPtr = std::shared_ptr<basicCancelRequest>;
    // 这个信道为发布者和消费者共用的类
    class Channel
    {
    public:
        using ptr = std::shared_ptr<Channel>;
        Channel(const std::string &id,
                const VirtualHost::ptr &host,
                const ConsumerManager::ptr &cmp,
                const ProtobufCodecPtr &codec,
                const muduo::net::TcpConnectionPtr &conn,
                const ThreadPool::ptr &pool) : _cid(id),
                                               _host(host),
                                               _cmp(cmp),
                                               _codec(codec),
                                               _conn(conn),
                                               _pool(pool)
        {
            LOG(INFO) << "new Channel: " << this << std::endl;
        }
        // 析构消费者
        ~Channel()
        {
            if (_consumer.get() != nullptr)
            {
                _cmp->remove(_consumer->tag, _consumer->qname);
            }
        }
        // 声明/创建交换机
        void declareExchange(const declareExchangeRequestPtr &req)
        {
            bool ret = _host->declareExchange(req->exchange_name(),
                                              req->exchange_type(), req->durable(),
                                              req->auto_delete(), req->args());
            basicResponse(ret, req->rid(), req->cid()); // 构建响应并发送
        }
        // 删除交换机
        void deleteExchange(const deleteExchangeRequestPtr &req)
        {
            _host->deleteExchange(req->exchange_name());
            basicResponse(true, req->rid(), req->cid());
        }

        // 声明和删除队列
        void declareQueue(const declareQueueRequestPtr &req)
        {
            bool ret = _host->declareQueue(req->queue_name(),
                                           req->durable(), req->exclusive(),
                                           req->auto_delete(), req->args());
            if (ret == false)
            {
                basicResponse(false, req->rid(), req->cid());
            }
            _cmp->initQueueConsumer(req->queue_name()); // 初始化队列的消费者管理句柄, 便于通过接口对该队列进行订阅
            basicResponse(true, req->rid(), req->cid());
        }
        void deleteQueue(const deleteQueueRequestPtr &req)
        {
            _cmp->destroyQueueConsumer(req->queue_name());
            _host->deleteQueue(req->queue_name());
            basicResponse(true, req->rid(), req->cid());
        }

        // 队列的绑定与解除
        void queueBind(const queueBindRequestPtr &req)
        {
            bool ret = _host->Bind(req->exchange_name(),
                                   req->queue_name(), req->binding_key());
            basicResponse(ret, req->rid(), req->cid());
        }
        void queueUnBind(const queueUnBindRequestPtr &req)
        {
            _host->unBind(req->exchange_name(), req->queue_name());
            basicResponse(true, req->rid(), req->cid());
        }
        // 消息的发布
        void basicPublish(const basicPublishRequestPtr &req)
        {
            // 1.获取交换机, 判断交换机是否存在
            auto ep = _host->selectExchange(req->exchange_name());
            if (ep.get() == nullptr)
            {
                return basicResponse(false, req->rid(), req->cid());
            }
            // 2. 进行路由交换, 判断消息发送到交换机绑定的哪个队列中
            MsgQueueBindingMap mqbm = _host->ExchangeBindings(req->exchange_name());

            BasicProperties *properties = nullptr;
            std::string routing_key;
            // 有属性就获取
            if (req->has_properties())
            {
                properties = req->mutable_properties();
                routing_key = properties->routing_key();
            }
            for (auto &binding : mqbm)
            {
                if (Router::route(ep->type, routing_key, binding.second->binding_key))
                {
                    // 3. 将消息添加到队列中（添加消息的管理）
                    _host->basicPublish(binding.first, properties, req->body());
                    // 4. 向线程池中添加一个消息消费任务（向指定队列的订阅者去推送消息--线程池完成）
                    //  _pool->Push(&Channel::consume, this, binding.first); // 不能直接将类成员函数传递过去
                    //  成员函数需要一个类的实例（即对象）来调用, 线程池中没有类对象, 无法调用
                    auto task = std::bind(&Channel::consume, this, binding.first); // 生成一个新的可调用对象
                    _pool->Push(task);
                }
            }
            return basicResponse(true, req->rid(), req->cid());
        }
        // 订阅队列消息
        void basicConsume(const basicConsumeRequestPtr &req)
        {
            // 1.判断队列是否存在
            bool ret = _host->existsQueue(req->queue_name());
            if (ret == false)
            {
                return basicResponse(false, req->rid(), req->cid());
            }
            // 2.创建队列的消费者
            auto cb = std::bind(&Channel::callback, this, std::placeholders::_1,
                                std::placeholders::_2, std::placeholders::_3);
            // 记录消费者, 信道关闭时以便于删除消费者
            // 把创建的消费者加入到指定队列的消费者列表中
            _consumer = _cmp->create(req->consumer_tag(), req->queue_name(), req->auto_ack(), cb);
            basicResponse(true, req->rid(), req->cid());
        }
        // 取消订阅
        void basicCancel(const basicCancelRequestPtr &req)
        {
            _cmp->remove(req->consumer_tag(), req->queue_name());
            return basicResponse(true, req->rid(), req->cid());
        }
        // 消息确认
        void basicAck(const basicAckRequestPtr &req)
        {
            _host->basicAck(req->queue_name(), req->message_id());
            return basicResponse(true, req->rid(), req->cid());
        }

    private:
        // 给消费客户端发送消息, 实现消息推送(将消息发送给自己)
        void callback(const std::string tag, const BasicProperties *bp, const std::string &body)
        {
            // 构建推送消息请求, 并将消息推送给消费者客户端
            basicConsumeResponse resp;
            resp.set_cid(_cid);
            resp.set_body(body);
            resp.set_consumer_tag(tag);
            // 属性不为空就设置
            if (bp != nullptr)
            {
                resp.mutable_properties()->set_id(bp->id());
                resp.mutable_properties()->set_delivery_mode(bp->delivery_mode());
                resp.mutable_properties()->set_routing_key(bp->routing_key());
            }
            
            _codec->send(_conn, resp);
        }
        void consume(const std::string &qname)
        {
            // 1. 从队列中取出消息
            MessagePtr mp = _host->basicConsume(qname);
            if (mp.get() == nullptr)
            {
                LOG(ERROR) << "执行执行消费任务失败，" << qname << " 队列没有消息!\n";
                return;
            }
            // 2. 从队列订阅者中取出一个订阅者
            Consumer::ptr cp = _cmp->choose(qname);
            if (cp.get() == nullptr)
            {
                LOG(ERROR) << "执行执行消费任务失败，" << qname << " 队列没有消费者!\n";
                return;
            }
            // 3. 调用订阅者对应的消息处理函数，实现消息的推送
            // 订阅队列创建消费者时注册
            // std::cout << "消息推送: " << cp->tag << ", " << mp->payload().body() << std::endl;
            cp->callback(cp->tag, mp->mutable_payload()->mutable_properties(), mp->payload().body());
            // 4. 判断如果订阅者是自动确认---不需要等待确认，直接删除消息，否则需要外部收到消息确认后再删除
            if (cp->auto_ack) _host->basicAck(qname, mp->payload().properties().id());
        }
        // 给发布客户端进行响应, 是否发布成功
        void basicResponse(bool ok, const std::string &rid, const std::string &cid)
        {
            basicCommonResponse resp;
            resp.set_rid(rid);
            resp.set_cid(cid);
            resp.set_ok(ok);
            _codec->send(_conn, resp);
        }

    private:
        // 信道与消费者一一对应
        // 消费者可能订阅多个不同的队列, 在信道管理的角度:
        // 遍历一个队列的消费者(列表vector),进行消息推送, 一个消费者可以出现在多个队列的消费者(列表);
        std::string _cid; // 信道标识id
        Consumer::ptr _consumer;
        muduo::net::TcpConnectionPtr _conn;
        ProtobufCodecPtr _codec; // 协议处理器

        ConsumerManager::ptr _cmp; // 队列消费者的管理句柄, 消费者与队列管理一一对应
        VirtualHost::ptr _host;
        ThreadPool::ptr _pool;
    };
    // 信道的管理
    class ChannelManager
    {
    public:
        using ptr = std::shared_ptr<ChannelManager>;
        ChannelManager()
        {
        }
        // 打开信道(创建信道)
        bool openChannel(const std::string &id,
                    const VirtualHost::ptr &host,
                    const ConsumerManager::ptr &cmp,
                    const ProtobufCodecPtr &codec,
                    const muduo::net::TcpConnectionPtr &conn,
                    const ThreadPool::ptr &pool)
        {
            std::unique_lock<std::mutex> lock(_mtx);
            auto it = _channels.find(id);
            if(it != _channels.end())
            {
                LOG(INFO) << "信道已经存在!\n";
                return false;
            }
            auto channel = std::make_shared<Channel>(id, host, cmp, codec, conn, pool);
            _channels.insert(std::make_pair(id, channel));
            return true;
        }
        void closeChannel(const std::string &id)
        {
            std::unique_lock<std::mutex> lock(_mtx);
            _channels.erase(id);
        }
        Channel::ptr getChannel(const std::string &id)
        {
            std::unique_lock<std::mutex> lock(_mtx);
            auto it = _channels.find(id);
            if(it == _channels.end())
            {
                return Channel::ptr();
            }
            return it->second;
        }
    private:
        std::mutex _mtx;
        std::unordered_map<std::string, Channel::ptr> _channels;
    };
}
